package com.galeno.utils;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.galeno.pojo.UserBeanAndProvince;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.apache.http.HttpResponse;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
import org.apache.http.impl.nio.client.HttpAsyncClients;
import org.apache.http.util.EntityUtils;

import java.util.Collections;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Future;
import java.util.function.Supplier;

/**
 * @author galeno
 * @Title:
 * @Description:
 * @date 2021/10/2719:24
 */
public class AsyncHttpQueryFunction extends RichAsyncFunction<String, UserBeanAndProvince> {
    private String key = "53bf6a55e756ce44f316706349d65096";
    private transient CloseableHttpAsyncClient httpClient;

    @Override
    public void open(Configuration parameters) throws Exception {
        //创建异步查询的HTTPClient
        //创建一个异步的HttpClient连接池
        //初始化异步的HttpClient
        RequestConfig requestConfig = RequestConfig.custom()
                .setSocketTimeout(3000)
                .setConnectTimeout(3000)
                .build();
        httpClient = HttpAsyncClients.custom()
                .setMaxConnTotal(20)
                .setDefaultRequestConfig(requestConfig)
                .build();
        httpClient.start();


    }


    @Override
    public void close() throws Exception {
        httpClient.close();
    }

    @Override
    public void asyncInvoke(String line, ResultFuture<UserBeanAndProvince> resultFuture) throws Exception {
        try {
            UserBeanAndProvince userBeanAndProvince = JSON.parseObject(line, UserBeanAndProvince.class);

            //异步查询
            double longitude = Double.parseDouble(userBeanAndProvince.getLongitude());
            double latitude = Double.parseDouble(userBeanAndProvince.getLatitude());
            HttpGet httpGet = new HttpGet("https://restapi.amap.com/v3/geocode/regeo?&location="+ longitude+"," + latitude +"&key=" + key);
            //查询返回Future
            Future<HttpResponse> future = httpClient.execute(httpGet, null);

            //从Future中取数据
            CompletableFuture.supplyAsync(new Supplier<UserBeanAndProvince>() {

                @Override
                public UserBeanAndProvince get() {
                    try {
                        HttpResponse response = future.get();
                        String province = null;
                        String city = null;
                        if (response.getStatusLine().getStatusCode() == 200) {
                            //获取请求的json字符串
                            String result = EntityUtils.toString(response.getEntity());
                            //System.out.println(result);
                            //转成json对象
                            JSONObject jsonObj = JSON.parseObject(result);
                            //获取位置信息
                            JSONObject regeocode = jsonObj.getJSONObject("regeocode");
                            if (regeocode != null && !regeocode.isEmpty()) {
                                JSONObject address = regeocode.getJSONObject("addressComponent");
                                //获取省市区
                                province = address.getString("province");
                                city = address.getString("city");
                                //String businessAreas = address.getString("businessAreas");
                            }
                        }
                        userBeanAndProvince.setProvince(province);
                      //  orderBean.province = province;
                        userBeanAndProvince.setCity(city);
                      //  orderBean.city = city;
                        return userBeanAndProvince;
                    } catch (Exception e) {
                        // Normally handled explicitly.
                        return null;
                    }
                }
            }).thenAccept((UserBeanAndProvince result) -> {
                resultFuture.complete(Collections.singleton(result));
            });

        } catch (Exception e) {
            resultFuture.complete(Collections.singleton(null));
        }

    }
}
